package com.cmiov.pulsar.pulsardemo;

import lombok.extern.slf4j.Slf4j;
import org.apache.pulsar.client.api.*;
import org.apache.pulsar.client.impl.schema.JSONSchema;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import java.util.Date;
import java.util.Random;

/**
 * * Licensed to CMDT,Inc. under the terms of the CMDT
 * Software License version 1.0.
 * <p>
 * See the NOTICE file distributed with this work for additional
 * information regarding copyright ownership.
 * ----------------------------------------------------------------------------
 * Date           Author               Version        Comments
 * 18-12-22       joe chow                 1.0            Initial Version
 */
@Service
@Slf4j
public class PulsarDemoService implements MessageListener<User>, ReaderListener<User> {

    private Consumer<User> consumer;

    private Producer<User> producer;

    private Reader<User> reader;

    private Random random = new Random(71);

    @PostConstruct
    private void init() {
        String localClusterUrl = "pulsar://localhost:6650";

        try {
            PulsarClient client = PulsarClient.builder().serviceUrl(localClusterUrl).build();
            consumer = client.newConsumer(JSONSchema.of(User.class))
                    .topic("user-topic")
                    .subscriptionName("joe-chow")
                    .messageListener(this)
                    .subscribe();
            producer = client.newProducer(JSONSchema.of(User.class)).topic("user-topic").create();

            reader = client.newReader(JSONSchema.of(User.class))
                    .topic("user-topic")
                    .readerListener(this)
                    .startMessageId(MessageId.earliest)
                    .create();
        } catch (PulsarClientException e) {
            e.printStackTrace();
        }


    }

    public void demo() {
        try {
            User user = new User();
            user.setName("joe");
            user.setAge(random.nextInt(100));
            user.setTime(new Date());
            producer.newMessage().value(user).send();
        } catch (PulsarClientException e) {
            e.printStackTrace();
        }
    }

    @Override
    public void received(Consumer<User> consumer, Message<User> message) {
        log.info("received pulsar message {} from topic {}",
                message.getValue(), message.getTopicName());
    }

    @Override
    public void reachedEndOfTopic(Consumer<User> consumer) {

    }

    @Override
    public void received(Reader<User> reader, Message<User> message) {
        log.info("reader received pulsar message {} from topic {}",
                message.getValue(), message.getTopicName());
    }
}
